package com.atguigu.flink.chapter7;

import com.atguigu.flink.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/19 14:56
 */
public class Flink07_Chapter7_WaterMark_Ordered {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> wms = WatermarkStrategy
          .<WaterSensor>forMonotonousTimestamps()
          .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
              @Override
              public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                  return element.getTs() * 1000;  //使用毫秒. 返回event time
              }
          });


        env
          .socketTextStream("hadoop162", 9999)
          .map(line -> {
              String[] split = line.split(",");
              return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
          })
          .assignTimestampsAndWatermarks(wms)
          .keyBy(WaterSensor::getId)
          .window(TumblingEventTimeWindows.of(Time.seconds(5)))
          .sum("vc")
          .print();

        env.execute();
    }
}
/*
sensor_1,5,10
sensor_1,6,10
sensor_1,7,10
 */
